package io.summer.gadget.channel.core.impl;

import io.summer.basis.api.ApiResponse;
import io.summer.gadget.channel.SummerChannel;
import io.summer.gadget.channel.core.ChannelRequest;
import io.summer.gadget.channel.core.MessageDispatcher;
import io.summer.gadget.channel.core.Protocol;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Collections;
import java.util.List;

/**
 * @author Moonlight
 */
@Slf4j
public final class MessageDispatcherImpl implements MessageDispatcher {
  private final SummerChannel summerChannel;
  private final Protocol protocol;

  public MessageDispatcherImpl(SummerChannel channel, Protocol protocol) {
    this.summerChannel = channel;
    this.protocol = protocol;
  }

  @SuppressWarnings("unchecked")
  private byte[] processRequest(ChannelRequest request) throws IOException {
    List<ChannelConsumer> consumers = summerChannel.getChannelConsumers(request.getTopic());
    if (consumers.isEmpty()) {
      log.warn("processRequest no consumers !!");
    }
    byte[] lastResult = null;
    for (ChannelConsumer consumer : consumers) {
      if (!consumer.check()) {
        log.warn("no valid consumer on request: {}", request);
        continue;
      }
      Object data = protocol.deserialize(request.getData(), consumer.getInputType());
      Object r = consumer.consumer.apply(data);
      if (consumer == consumers.get(consumers.size() - 1)) {
        lastResult = protocol.serialize(r);
      }
    }
    return lastResult;
  }

  @SneakyThrows(IOException.class)
  @Override
  public byte[] dispatch(ChannelRequest request) {
    if (log.isDebugEnabled()) {
      log.debug("receive message: {}", request);
    }
    return processRequest(request);
  }

  @Override
  @Async
  public byte[] dispatchAsync(ChannelRequest request) {
    if (log.isDebugEnabled()) {
      log.debug("receive message async: {}", request);
    }
    try {
      processRequest(request);
    } catch (IOException e) {
      log.warn("dispatchAsync failed: ", e);
    }
    return ApiResponse.ok().toSimpleString().getBytes();
  }
}
